package com.lds.memory.kafka.topic;

import com.lds.memory.kafka.enums.ErrorEnum;
import com.lds.memory.kafka.partition.Partition;
import com.lds.memory.kafka.util.AssertUtils;
import com.lds.memory.kafka.util.EnvUtils;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.List;

public final class TopicGenerator {
    private static final int PARTITION_COUNT_ENV = EnvUtils.getInt("MEMORY_KAFKA.PARTITION_COUNT_ENV");
    private static final int MIN_PARTITION_COUNT = 1;
    private static final int MAX_PARTITION_COUNT = Runtime.getRuntime().availableProcessors();

    private static final int CAPACITY_OF_PARTITION_ENV = EnvUtils.getInt("MEMORY_KAFKA.CAPACITY_OF_PARTITION_ENV");
    private static final int MIN_CAPACITY_OF_PARTITION = 100;
    private static final int MAX_CAPACITY_OF_PARTITION = 1000;

    private TopicGenerator() {
    }

    public static Topic generate(TopicConfig topicConfig) {
        AssertUtils.assertTrue(StringUtils.isNotEmpty(topicConfig.getTopicName()), ErrorEnum.TOPIC_NAME_IS_EMPTY);

        int validPartitionCount = buildPartitionCount(topicConfig.getPartitionCount());
        int validCapacityOfPartition = buildCapacityOfPartition(topicConfig.getCapacityOfPartition());

        List<Partition> partitions = new ArrayList<>();
        for (int i = 0; i < validPartitionCount; i++) {
            partitions.add(new Partition(validCapacityOfPartition, topicConfig.getRejectedStrategy()));
        }
        return new Topic(topicConfig.getTopicName(), partitions);
    }

    private static int buildPartitionCount(int partitionCount) {
        if (partitionCount == 0) partitionCount = PARTITION_COUNT_ENV;
        if (partitionCount < MIN_PARTITION_COUNT) partitionCount = MIN_PARTITION_COUNT;
        if (partitionCount > MAX_PARTITION_COUNT) partitionCount = MAX_PARTITION_COUNT;
        return partitionCount;
    }

    private static int buildCapacityOfPartition(int capacityOfPartition) {
        if (capacityOfPartition == 0) capacityOfPartition = CAPACITY_OF_PARTITION_ENV;
        if (capacityOfPartition < MIN_CAPACITY_OF_PARTITION) capacityOfPartition = MIN_CAPACITY_OF_PARTITION;
        if (capacityOfPartition > MAX_CAPACITY_OF_PARTITION) capacityOfPartition = MAX_CAPACITY_OF_PARTITION;
        return capacityOfPartition;
    }

}
